#include "KvServer.h"
#include <mprpcprovider.h>  // RPC服务提供者（用于注册RPC接口）
#include "mprpcconfig.h"    // RPC配置解析（读取节点IP、端口等）

// 构造函数：初始化KV服务器（Raft节点、RPC服务、存储等）
KvServer::KvServer(int me, int maxraftstate, std::string nodeInforFileName, short port) 
    :   m_me(me),
        m_skipList(6),
        m_maxRaftState(maxraftstate),
        m_lastSnapShotRaftLogIndex(0) {  // 跳表初始化（层级=6）

    // 1. 初始化持久化工具
    std::shared_ptr<raftCore::Persister> persister = std::make_shared<raftCore::Persister>(me);

    // 2. 初始化与Raft的通信管道
    applyChan = std::make_shared<common::LockQueue<raftCore::ApplyMsg> >();  // 初始化与Raft的通信管道

    // 3. 初始化Raft节点
    m_raftNode = std::make_shared<raftCore::Raft>();

    // 4. 启动RPC服务线程
    std::thread t([this, port]() -> void {
        rpc::MprpcProvider provider;  // RPC服务提供者
        provider.NotifyService(this);  // 注册KV服务器的RPC接口
        provider.NotifyService(m_raftNode.get());  // 注册Raft节点的RPC接口（用于节点间通信）
        provider.Run(m_me, port);  // 启动RPC服务（监听指定端口）
    });
    t.detach();  // 分离线程（后台运行）

    // 5. 等待其他节点启动（简单同步机制）
    std::cout << "raftServer node:" << m_me << " start to sleep to wait all ohter raftnode start!!!!" << std::endl;
    sleep(6);
    std::cout << "raftServer node:" << m_me << " wake up!!!! start to connect other raftnode" << std::endl;

    // 6. 读取集群节点配置
    rpc::MprpcConfig config;
    config.LoadConfig(nodeInforFileName.c_str());
    std::vector<std::pair<std::string, short> > ipPortVt;
    for (int i = 0; i < INT_MAX - 1; ++i) {
        std::string node = "node" + std::to_string(i);
        std::string nodeIp = config.Load(node + "ip");
        std::string nodePortStr = config.Load(node + "port");
        if (nodeIp.empty()) {  // 无更多节点配置
            break;
        }
        ipPortVt.emplace_back(nodeIp, atoi(nodePortStr.c_str()));  // 存储节点地址
    }

    // 7. 连接集群中其他节点
    std::vector<std::shared_ptr<raftCore::RaftRpcUtil> > servers;
    for (int i = 0; i < ipPortVt.size(); ++i) {
        if (i == m_me) {  // 跳过自身
            servers.push_back(nullptr);
            continue;
        }
        std::string otherNodeIp = ipPortVt[i].first;
        short otherNodePort = ipPortVt[i].second;
        auto *rpc = new raftCore::RaftRpcUtil(otherNodeIp, otherNodePort);  // 创建RPC工具
        servers.push_back(std::shared_ptr<raftCore::RaftRpcUtil>(rpc));   // 管理RPC工具
        std::cout << "node" << m_me << " 连接node" << i << "success!" << std::endl;
    }

    // 8. 等待节点间连接完成
    sleep(ipPortVt.size() - me);

    // 9. 初始化Raft节点
    m_raftNode->init(servers, m_me, persister, applyChan);

    // 10. 从快照恢复KV状态
    auto snapshot = persister->ReadSnapshot();  // 读取已保存的快照
    if (!snapshot.empty()) {
        ReadSnapShotToInstall(snapshot);  // 恢复快照
    }

    // 11. 启动日志处理循环（阻塞当前线程）
    // 让构造函数的 “初始化逻辑” 与 “日志处理逻辑” 分离
    std::thread t2(&KvServer::ReadRaftApplyCommandLoop, this);
    t2.join();  // 确保 KV 服务进程不退出，持续提供服务
    // 让构造函数执行完毕，外部代码获取kvserver实例
}

// 调试用：打印当前KV数据库内容（仅在Debug模式下生效）
void KvServer::DprintfKVDB() {
    if (!common::DEBUG) {  // 全局Debug开关控制
        return;
    }
    std::lock_guard<std::mutex> lg(m_mtx);  // 加锁保证线程安全
    common::DEFER [&]() {  // 延迟执行块（可能是自定义宏，确保打印逻辑执行）
        m_skipList.display_list();  // 打印跳表内容（核心存储）
    };
}

// 执行Append操作：向指定键追加值（应用到本地KV存储）
void KvServer::ExecuteAppendOpOnKVDB(common::Op op) {
    m_mtx.lock();  // 加锁保护共享数据

    // 使用跳表插入/更新键值（Append逻辑：若键存在则追加，否则插入）
    m_skipList.insert_set_element(op.key, op.value);

    // 更新客户端最后请求ID（防止重复执行）
    m_lastRequestId[op.client_id] = op.request_id;
    m_mtx.unlock();  // 解锁

    DprintfKVDB();  // 调试打印
}

// 执行Get操作：查询指定键的值（从本地KV存储读取）
void KvServer::ExecuteGetOpOnKVDB(common::Op op, std::string *value, bool *exist) {
    m_mtx.lock();  // 加锁
    *value = "";   // 初始化输出参数
    *exist = false;
    // 从跳表查询键，若存在则赋值value并标记exist为true
    if (m_skipList.search_element(op.key, *value)) {
        *exist = true;
    }

    // 更新客户端最后请求ID
    m_lastRequestId[op.client_id] = op.request_id;
    m_mtx.unlock();  // 解锁

    DprintfKVDB();  // 调试打印
}

// 执行Put操作：设置指定键的值（覆盖式更新）
void KvServer::ExecutePutOpOnKVDB(common::Op op) {
    m_mtx.lock();  // 加锁
    // 跳表插入/覆盖键值
    m_skipList.insert_set_element(op.key, op.value);

    // 更新客户端最后请求ID
    m_lastRequestId[op.client_id] = op.request_id;
    m_mtx.unlock();  // 解锁

    DprintfKVDB();  // 调试打印
}

// 处理客户端Get请求（内部逻辑，非RPC接口）
void KvServer::Get(const raftKVRpcProctoc::GetArgs *args, raftKVRpcProctoc::GetReply *reply) {
    // 封装客户端请求为Op结构（Raft日志条目）
    common::Op op;
    op.operation = "Get";
    op.key = args->key();
    op.value = "";  // Get操作无需Value
    op.client_id = args->clientid();
    op.request_id = args->requestid();

    int raftIndex = -1;  // 该操作在Raft日志中的索引
    int term = -1;          // 未使用（可能是任期）
    bool isLeader = false;  // 当前节点是否为Raft Leader

    // 将操作提交到Raft（Start方法：发起共识，返回日志索引和是否为Leader）
    m_raftNode->Start(op, &raftIndex, &term, &isLeader);

    if (!isLeader) {  // 若不是Leader，返回错误（客户端需重试其他节点）
        reply->set_err(common::ErrWrongLeader);
        return;
    }

    // 创建等待队列：用于等待Raft将该日志提交后通知当前请求
    m_mtx.lock();
    if (waitApplyCh.find(raftIndex) == waitApplyCh.end()) {
        // 为该日志索引创建新的等待队列
        waitApplyCh.insert(std::make_pair(raftIndex, new common::LockQueue<common::Op>()));
    }
    auto chForRaftIndex = waitApplyCh[raftIndex];  // 获取当前日志索引的等待队列
    m_mtx.unlock();  // 尽早解锁，避免阻塞

    // 等待Raft提交该操作（超时时间由CONSENSUS_TIMEOUT定义）
    common::Op raftCommitOp;  // 用于接收Raft提交的操作
    if (!chForRaftIndex->TimeoutPop(common::CONSENSUS_TIMEOUT, &raftCommitOp)) {
        // 超时处理：Raft未在规定时间内提交该日志
        int term = -1;
        bool isLeader = false;
        m_raftNode->GetState(&term, &isLeader);  // 再次检查是否为Leader

        // 若请求已重复（客户端重试）且仍为Leader，则直接执行（Get操作可重复）
        if (ifRequestDuplicate(op.client_id, op.request_id) && isLeader) {
            std::string value;
            bool exist = false;
            ExecuteGetOpOnKVDB(op, &value, &exist);  // 直接查询本地存储
            if (exist) {
                reply->set_err(common::OK);
                reply->set_value(value);
            } 
            else {
                reply->set_err(common::ErrNoKey);
            }
        } 
        else {
            // 非重复请求或已不是Leader，返回错误（客户端重试）
            reply->set_err(common::ErrWrongLeader);
        }
    } 
    else {
        // 成功收到Raft提交的操作：验证操作匹配性（防止Leader变更导致的日志覆盖）
        if (raftCommitOp.client_id == op.client_id && raftCommitOp.request_id== op.request_id) {
            // 操作匹配：执行Get并返回结果
            std::string value;
            bool exist = false;
            ExecuteGetOpOnKVDB(op, &value, &exist);
            if (exist) {
                reply->set_err(common::OK);
                reply->set_value(value);
            } 
            else {
                reply->set_err(common::ErrNoKey);
            }
        } 
        else {
            // 操作不匹配（可能Leader变更），返回错误
            reply->set_err(common::ErrWrongLeader);
        }
    }

    // 清理：删除该日志索引的等待队列（避免内存泄漏）
    m_mtx.lock();
    auto tmp = waitApplyCh[raftIndex];
    waitApplyCh.erase(raftIndex);
    delete tmp;
    m_mtx.unlock();
}

// 处理从Raft传来的已提交日志（应用到本地KV存储）
void KvServer::GetCommandFromRaft(raftCore::ApplyMsg message) {
    common::Op op;
    op.ParseFromString(message.Command);  // 从Raft消息中解析操作

    // 调试日志：打印收到的操作详情
    common::DPrintf(
        "[KvServer::GetCommandFromRaft-kvserver{%d}] , Got Command --> Index:{%d} , ClientId {%s}, RequestId {%d}, "
        "Opreation {%s}, Key :{%s}, Value :{%s}",
        m_me, message.CommandIndex, &op.client_id, op.request_id, &op.operation, &op.key, &op.value);

    // 忽略已被快照覆盖的旧日志（日志索引 <= 最后快照索引）
    if (message.CommandIndex <= m_lastSnapShotRaftLogIndex) {
        return;
    }

    // 状态机应用：仅处理非重复的操作（保证幂等性）
    if (!ifRequestDuplicate(op.client_id, op.request_id)) {
        // 根据操作类型执行对应KV操作
        if (op.operation == "Put") {
            ExecutePutOpOnKVDB(op);
        }
        if (op.operation == "Append") {
            ExecuteAppendOpOnKVDB(op);
        }
        // Get操作无需在此执行（Get由客户端请求时直接查询，不修改状态）
    }

    // 检查是否需要生成快照（当Raft日志大小超过阈值时）
    if (m_maxRaftState != -1) {
        IfNeedToSendSnapShotCommand(message.CommandIndex, 10);
    }

    // 通知等待队列：该操作已提交（唤醒等待的客户端请求）
    SendMessageToWaitChan(op, message.CommandIndex);
}

// 检查客户端请求是否重复（根据ClientId和RequestId）
bool KvServer::ifRequestDuplicate(std::string ClientId, int RequestId) {
    std::lock_guard<std::mutex> lg(m_mtx);  // 加锁
    // 若客户端未记录或请求ID大于最后一次ID，则不重复
    if (m_lastRequestId.find(ClientId) == m_lastRequestId.end()) {
        return false;
    }
    // 请求ID <= 最后一次ID：重复
    return RequestId <= m_lastRequestId[ClientId];
}

// 处理客户端Put/Append请求（内部逻辑，非RPC接口）
void KvServer::PutAppend(const raftKVRpcProctoc::PutAppendArgs *args, raftKVRpcProctoc::PutAppendReply *reply) {
    // 封装客户端请求为Op结构
    common::Op op;
    op.operation = args->op();  // "Put"或"Append"
    op.key = args->key();
    op.value = args->value();
    op.client_id = args->clientid();
    op.request_id = args->requestid();

    int raftIndex = -1;
    int _ = -1;
    bool isleader = false;

    // 提交操作到Raft
    m_raftNode->Start(op, &raftIndex, &_, &isleader);

    if (!isleader) {  // 非Leader，返回错误
        common::DPrintf(
            "[func -KvServer::PutAppend -kvserver{%d}]From Client %s (Request %d) To Server %d, key %s, raftIndex %d , but "
            "not leader",
            m_me, &args->clientid(), args->requestid(), m_me, &op.key, raftIndex);
        reply->set_err(common::ErrWrongLeader);
        return;
    }

    // 创建等待队列（同Get逻辑）
    m_mtx.lock();
    if (waitApplyCh.find(raftIndex) == waitApplyCh.end()) {
        waitApplyCh.insert(std::make_pair(raftIndex, new common::LockQueue<common::Op>()));
    }
    auto chForRaftIndex = waitApplyCh[raftIndex];
    m_mtx.unlock();

    // 等待Raft提交操作
    common::Op raftCommitOp;
    if (!chForRaftIndex->TimeoutPop(common::CONSENSUS_TIMEOUT, &raftCommitOp)) {
        // 超时处理：若请求重复，返回成功（Put/Append需保证幂等性）
        if (ifRequestDuplicate(op.client_id, op.request_id)) {
            reply->set_err(common::OK);
        } 
        else {
            // 非重复请求，返回错误（客户端重试）
            reply->set_err(common::ErrWrongLeader);
        }
    } 
    else {
        // 验证操作匹配性（防止Leader变更）
        if (raftCommitOp.client_id == op.client_id && raftCommitOp.request_id== op.request_id) {
            reply->set_err(common::OK);  // 操作成功提交
        } 
        else {
            reply->set_err(common::ErrWrongLeader);
        }
    }

    // 清理等待队列
    m_mtx.lock();
    auto tmp = waitApplyCh[raftIndex];
    waitApplyCh.erase(raftIndex);
    delete tmp;
    m_mtx.unlock();
}

// 循环读取Raft提交的日志并应用（核心工作循环）
void KvServer::ReadRaftApplyCommandLoop() {
    while (true) {  // 无限循环，持续处理
        auto message = applyChan->Pop();  // 从管道阻塞读取Raft消息（已提交的日志）
        common::DPrintf(
            "---------------tmp-------------[func-KvServer::ReadRaftApplyCommandLoop()-kvserver{%d}] 收到了下raft的消息",
            m_me);

        // 处理正常日志条目（CommandValid为true）
        if (message.CommandValid) {
            GetCommandFromRaft(message);  // 应用到本地KV存储
        }
        // 处理快照消息（SnapshotValid为true）
        if (message.SnapshotValid) {
            GetSnapShotFromRaft(message);  // 安装快照
        }
    }
}

// 从快照数据恢复KV状态（用于节点启动或同步）
void KvServer::ReadSnapShotToInstall(std::string snapshot) {
    if (snapshot.empty()) {  // 空快照：初始化状态
        return;
    }
    parseFromString(snapshot);  // 反序列化快照数据（恢复跳表和请求ID记录）
}

// 向等待队列发送消息（通知客户端请求：操作已提交）
bool KvServer::SendMessageToWaitChan(const common::Op &op, int raftIndex) {
    std::lock_guard<std::mutex> lg(m_mtx);  // 加锁
    common::DPrintf(
        "[RaftApplyMessageSendToWaitChan--> raftserver{%d}] , Send Command --> Index:{%d} , ClientId {%d}, RequestId "
        "{%d}, Opreation {%v}, Key :{%v}, Value :{%v}",
        m_me, raftIndex, &op.client_id, op.request_id, &op.operation, &op.key, &op.value);

    // 若等待队列不存在（可能已超时清理），返回失败
    if (waitApplyCh.find(raftIndex) == waitApplyCh.end()) {
        return false;
    }
    // 向队列推送操作，唤醒等待的客户端请求
    waitApplyCh[raftIndex]->Push(op);
    return true;
}

// 检查是否需要生成快照（Raft日志过大时）
void KvServer::IfNeedToSendSnapShotCommand(int raftIndex, int proportion) {
    // 若Raft日志大小超过阈值的1/10，触发快照
    if (m_raftNode->GetRaftStateSize() > m_maxRaftState / proportion) {
        auto snapshot = MakeSnapShot();  // 生成当前状态的快照
        m_raftNode->Snapshot(raftIndex, snapshot);  // 通知Raft保存快照
    }
}

// 处理Raft传来的快照（如Leader同步给落后节点的快照）
void KvServer::GetSnapShotFromRaft(raftCore::ApplyMsg message) {
    std::lock_guard<std::mutex> lg(m_mtx);  // 加锁

    // 验证快照合法性（任期和索引），合法则安装
    if (m_raftNode->CondInstallSnapshot(message.SnapshotTerm, message.SnapshotIndex, message.Snapshot)) {
        ReadSnapShotToInstall(message.Snapshot);  // 恢复快照数据
        m_lastSnapShotRaftLogIndex = message.SnapshotIndex;  // 更新最后快照索引
    }
}

// 生成当前KV状态的快照（序列化存储）
std::string KvServer::MakeSnapShot() {
    std::lock_guard<std::mutex> lg(m_mtx);  // 加锁保证状态一致性
    std::string snapshotData = getSnapshotData();  // 序列化跳表和请求ID记录
    return snapshotData;
}

// RPC接口实现：PutAppend（客户端远程调用入口）
void KvServer::PutAppend(google::protobuf::RpcController *controller, const ::raftKVRpcProctoc::PutAppendArgs *request,
                         ::raftKVRpcProctoc::PutAppendReply *response, ::google::protobuf::Closure *done) {
    KvServer::PutAppend(request, response);  // 调用内部处理逻辑
    done->Run();  // 通知RPC框架：处理完成，发送响应
}

// RPC接口实现：Get（客户端远程调用入口）
void KvServer::Get(google::protobuf::RpcController *controller, const ::raftKVRpcProctoc::GetArgs *request,
                   ::raftKVRpcProctoc::GetReply *response, ::google::protobuf::Closure *done) {
    KvServer::Get(request, response);  // 调用内部处理逻辑
    done->Run();  // 通知RPC框架：处理完成
}